package mqtt_thingsboard_connector

import (
	"encoding/json"
	"fmt"
	"github.com/eclipse/paho.mqtt.golang"
	"github.com/panjf2000/ants/v2"
	"github.com/sirupsen/logrus"
	"go-iot-gateway/internal/apk/maths"
	"go-iot-gateway/internal/channel"
	mqtt_connector "go-iot-gateway/internal/channel/connector/mqtt"
	channel_define "go-iot-gateway/internal/channel/define"
	"go-iot-gateway/internal/conf_table"
	"go-iot-gateway/internal/conf_table/define"
	"go-iot-gateway/internal/storage"
	storage_define "go-iot-gateway/internal/storage/define"
	"strings"
	"sync"
	"time"
)

const (
	TOPIC_CONNECT             = "connect"
	TOPIC_DISCONNECT          = "disconnect"
	TOPIC_RPC                 = "rpc"
	TOPIC_TELEMETRY           = "telemetry"
	TOPIC_ATTRIBUTES          = "attributes"
	TOPIC_ATTRIBUTES_REQUEST  = "attributes/request"
	TOPIC_ATTRIBUTES_RESPONSE = "attributes/response"
	TOPIC_CLAIMING            = "claim"
)

// mqtt 客户参考手册  https://www.jianshu.com/p/c23724b7e14f
type ThingsBoardMqttClientChannel struct {
	protocolInfo         *ThingsBoardMqttProtocolInfo
	ChannelInfo          channel_define.ChannelInfo
	mqttClient           mqtt.Client
	isClosed             bool
	channelManager       *channel.ChannelManager
	readTagGroupByDevice map[string][]*conf_table_define.TagInfo //读点位 key: 设备 | val: 转发点
	//writeTagList         []*conf_table_define.TagInfo            //写点位
	waterlineFlags map[string]int64 //发布数据的水位线,记录的是每个设备最后上传数据的时间
	logger         *logrus.Logger
	rcpChan        chan *RPCRequest //rpc 队列
	rcpTaskPool    *ants.Pool       //rpc 任务池
	connectLock    sync.Mutex       //mqtt 连接锁
}

func (t *ThingsBoardMqttClientChannel) Open() {
	t.logger.Debug("ThingsBoardMqttClientChannel 打开")
	//todo 加锁，只调用一次
	//todo 资料初始化应该放在这里，如通道 ， 线程池 ，连接...
	mqtt.CRITICAL = &mqtt_connector.MqttLogger{t.logger, logrus.ErrorLevel} //logrus.FatalLevel
	mqtt.ERROR = &mqtt_connector.MqttLogger{t.logger, logrus.ErrorLevel}
	mqtt.WARN = &mqtt_connector.MqttLogger{t.logger, logrus.WarnLevel}
	//mqtt.DEBUG = &mqtt_connector.MqttLogger{t.logger, logrus.DebugLevel} //todo debug 日志打开

	t.isClosed = false
	mattOpts := mqtt.NewClientOptions()

	//tcp://broker.emqx.io:1883
	mattOpts.AddBroker(fmt.Sprintf("tcp://%s:%d", t.ChannelInfo.ConnectParams.IpAddr, t.ChannelInfo.ConnectParams.Port))
	mattOpts.SetClientID(t.protocolInfo.ClientId())
	mattOpts.SetUsername(t.protocolInfo.UserName)
	mattOpts.SetPassword(t.protocolInfo.Password)
	mattOpts.SetMaxReconnectInterval(time.Duration(7) * time.Second)
	mattOpts.SetConnectTimeout(time.Duration(8) * time.Second)
	mattOpts.SetWriteTimeout(time.Duration(7) * time.Second)
	mattOpts.SetPingTimeout(time.Duration(7) * time.Second)
	mattOpts.OnConnect = t.onConnect
	mattOpts.OnConnectionLost = t.onConnectionLost
	t.mqttClient = mqtt.NewClient(mattOpts)

	t.ChannelInfo.SetStatus(channel_define.CONNECT_STATUS_CONNECTING, "连接中")
	go func() {
		t.doConnect()
	}()

	go func() {
		t.rpcHandle()
	}()

	deviceConfInfos := conf_table.FindAllDeviceByChlId(t.ChannelInfo.ChlId)
	if len(deviceConfInfos) == 0 {
		return
	}

	//t.writeTagList = make([]*conf_table_define.TagInfo, 0, 30)
	t.readTagGroupByDevice = make(map[string][]*conf_table_define.TagInfo, len(deviceConfInfos))
	t.waterlineFlags = make(map[string]int64, len(deviceConfInfos))
	for _, deviceConfInfo := range deviceConfInfos {
		if !deviceConfInfo.IsValid() {
			continue
		}
		tagsList := make([]*conf_table_define.TagInfo, 0, 100)

		tagConfInfos := conf_table.FindAllTagByDeviceId(deviceConfInfo.Dev_id)
		if len(tagConfInfos) == 0 {
			continue
		}

		for _, tag := range tagConfInfos {
			// 查询原设备
			oriTag, _ := conf_table.FindAllTagByTagId(tag.Ori_tag_id)
			if oriTag == nil || !oriTag.DeviceInfo.IsValid() { // 原设备不可用
				continue
			}

			//switch tag.Collect_type {
			//case 1: //模拟量
			//	tagsList = append(tagsList, tag)
			//case 2: //开关量 0;1
			//	t.writeTagList = append(t.writeTagList, tag)
			//}

			if tag.Tag_type == conf_table_define.TAG_TYPE_YX || tag.Tag_type == conf_table_define.TAG_TYPE_YC || tag.Tag_type == conf_table_define.TAG_TYPE_YM {
				tagsList = append(tagsList, tag)
			} else if tag.Tag_type == conf_table_define.TAG_TYPE_YT || tag.Tag_type == conf_table_define.TAG_TYPE_YK {
				//t.writeTagList = append(t.writeTagList, tag)
			} else {
				//panic("标签类型匹配不到")
			}

		}

		t.readTagGroupByDevice[deviceConfInfo.Dev_id] = tagsList
		t.waterlineFlags[deviceConfInfo.Dev_id] = 0
	}

	go func() {
		t.push()
	}()

}

func (t *ThingsBoardMqttClientChannel) Close() {
	t.isClosed = true
	t.ChannelInfo.SetStatus(channel_define.CONNECT_STATUS_CLOSED, "已经关闭")
	t.mqttClient.Disconnect(1)
	close(t.rcpChan)
	t.rcpTaskPool.Release()
}

func (t *ThingsBoardMqttClientChannel) Info() channel_define.ChannelInfo {
	return t.ChannelInfo
}

func (t *ThingsBoardMqttClientChannel) doConnect() {
	//todo 查看连接链路
	if t.mqttClient.IsConnected() {
		t.ChannelInfo.SetStatus(channel_define.CONNECT_STATUS_CONNECTED, "已连接")
		return
	}

	getLock := t.connectLock.TryLock()

	if !getLock {
		return //没有拿到锁,说明已经进行连接了
	}

	defer t.connectLock.Unlock()

	if t.mqttClient.IsConnected() {
		t.ChannelInfo.SetStatus(channel_define.CONNECT_STATUS_CONNECTED, "已连接")
		return
	}

	for {
		t.ChannelInfo.SetStatus(channel_define.CONNECT_STATUS_CONNECTING, "连接中")
		token := t.mqttClient.Connect()

		if token.Wait() && token.Error() != nil {
			t.logger.Warn("mqtt 连接失败", token.Error())
			time.Sleep(7 * time.Second)
			continue
		} else {
			break
		}
	}

	t.ChannelInfo.SetStatus(channel_define.CONNECT_STATUS_CONNECTED, "已连接")
}

// 连接成功
func (t *ThingsBoardMqttClientChannel) onConnect(c mqtt.Client) {
	t.logger.Debug("mqtt 连接成功")
	t.ChannelInfo.SetStatus(channel_define.CONNECT_STATUS_CONNECTED, "已连接")
	if token := t.mqttClient.Subscribe(t.protocolInfo.SubscribeTopic+TOPIC_ATTRIBUTES, byte(1), func(c mqtt.Client, msg mqtt.Message) {
		t.onSubscribe(TOPIC_ATTRIBUTES, msg)
	}); token.Wait() && token.Error() != nil {
		//err(token.Error())
		t.logger.Errorln(TOPIC_ATTRIBUTES, "  mqtt 订阅失败")
	}
	if token := t.mqttClient.Subscribe(t.protocolInfo.SubscribeTopic+TOPIC_ATTRIBUTES_RESPONSE, byte(1), func(c mqtt.Client, msg mqtt.Message) {
		t.onSubscribe(TOPIC_ATTRIBUTES_RESPONSE, msg)
	}); token.Wait() && token.Error() != nil {
		t.logger.Errorln(TOPIC_ATTRIBUTES_RESPONSE, " mqtt 订阅失败")
	}
	if token := t.mqttClient.Subscribe(t.protocolInfo.SubscribeTopic+TOPIC_RPC, byte(1), func(c mqtt.Client, msg mqtt.Message) {
		t.onSubscribe(TOPIC_RPC, msg)
	}); token.Wait() && token.Error() != nil {
		t.logger.Errorln(TOPIC_RPC, " mqtt 订阅失败")
	}

}

// 断开连接
func (t *ThingsBoardMqttClientChannel) onConnectionLost(c mqtt.Client, err error) {
	t.ChannelInfo.SetStatus(channel_define.CONNECT_STATUS_ERROR, "连接错误")
	time.Sleep(5 * time.Second)
	if !c.IsConnected() {
		t.logger.Info("mqtt 断开连接,进行重连)")
		go func() {
			t.doConnect()
		}()
	}
}

func (t *ThingsBoardMqttClientChannel) onSubscribe(topic string, msg mqtt.Message) {
	t.logger.Debugln("[rpc-topic] ", topic)
	switch topic {
	case TOPIC_RPC:
		payload := msg.Payload()
		t.logger.Infof("[rcp-请求] %s \n", string(payload))
		rpcRequest := &RPCRequest{}

		err := json.Unmarshal(payload, rpcRequest)
		if err != nil {
			t.logger.Errorln("[rpc] 报文解析失败 err ", err)
			return
		}

		select {
		case t.rcpChan <- rpcRequest:
		case <-time.After(time.Second * 2): // 2秒超时
			t.logger.Warnln("rpc 写通道已经满了,消息将丢弃, 通道长度为 ", len(t.rcpChan))
			t.mqttRpcResponse(RPCResponseFailMsg(rpcRequest, fmt.Sprintf("[请求繁忙]当前任务数 %d ", len(t.rcpChan))))
		}
	default:
		payload := string(msg.Payload())
		t.logger.Infof("[onSubscribe] %s \n", payload)
	}

}

// 推送数据
func (t *ThingsBoardMqttClientChannel) push() {

	//交互频率
	frequency := maths.Max4int(100, t.ChannelInfo.Frequency)
	//每帧间隔
	waitTime := time.Duration(maths.Max4int(100, t.ChannelInfo.WaitTime))

	for !t.isClosed {

		//通道还没初始化好
		if t.mqttClient == nil || !t.mqttClient.IsConnected() {
			time.Sleep(waitTime * time.Millisecond)
			continue
		}

		beginTime := time.Now().UnixMilli()
		for devId, tags := range t.readTagGroupByDevice {
			if t.isClosed {
				return
			}

			if t.mqttClient == nil || !t.mqttClient.IsConnected() {
				break
			}

			if len(tags) <= 0 {
				continue
			}

			waterline, err := t.publishTelemetry(devId, t.waterlineFlags[devId])
			if err != nil {
				t.logger.Warn("thingsBoard 数据发送失败", err)
			} else {
				t.waterlineFlags[devId] = waterline
			}
			if waitTime > 0 {
				time.Sleep(waitTime * time.Millisecond) //每帧间隔
			}
		}

		//交互频率
		consumeTime := time.Now().UnixMilli() - beginTime
		pauseTime := int64(frequency) - consumeTime
		if pauseTime > 0 {
			time.Sleep(time.Duration(pauseTime) * time.Millisecond)
		}

	}

}

func (t *ThingsBoardMqttClientChannel) publishTelemetry(devId string, ts int64) (int64, error) {

	defer func() {
		//todo 优化 panic
		if rerr := recover(); rerr != nil {
			t.logger.Panicf("[panic] publishTelemetry error %s \n", rerr)
		}
	}()

	tagInfos := t.readTagGroupByDevice[devId]
	tagNames := make([]string, 0, len(tagInfos))
	for _, info := range tagInfos {
		tagNames = append(tagNames, "'"+info.Tag_name+"'")
	}

	qosFilter := storage_define.QOS_GOOD
	if t.protocolInfo.EnableBadQos {
		qosFilter = storage_define.QOS_BAD
	}

	querySql := "select * from " + storage_define.LATEST_TABLE_NAME + " where ts > ? and tag_name in (%s) and qos <= ? order by ts asc"
	valObjects := storage.FindLatestBySQL(fmt.Sprintf(querySql, strings.Join(tagNames, ",")), ts, qosFilter)
	dataLen := len(valObjects)
	if valObjects == nil || dataLen <= 0 {
		return ts, nil
	}

	lastTs := valObjects[dataLen-1].Ts()
	resultMap := t.packThingBoardMessage(devId, valObjects)
	t.logger.Debug("【上报】", resultMap)
	payload, err := json.Marshal(resultMap)
	if err != nil {
		return ts, err
	}
	qos := byte(1) //0:发送就不管了，最多一次;1发送之后依赖MQTT规范，是否启动重传消息，所以至少一次;2:发送之后依赖MQTT消息机制，确保只有一次;
	//todo 如果数据太多，可能要分批
	t.mqttClient.Publish(t.protocolInfo.PublicTopic+TOPIC_TELEMETRY, qos, false, payload)
	//取最后一条返回
	return lastTs, nil
}

// tb 消息打包
func (t *ThingsBoardMqttClientChannel) packThingBoardMessage(devId string, valObjects []storage_define.ValObject) map[string]interface{} {
	deviceConfInfo, _ := conf_table.FindAllDeviceByDevId(devId)

	groupByTs := make(map[int64]map[string]interface{}, 7)

	for _, valObject := range valObjects {
		ts := valObject.Ts()
		arr, ok := groupByTs[ts]
		if !ok {
			arr = make(map[string]interface{}, 30)
			groupByTs[ts] = arr
		}
		tag, _ := conf_table.FindAllTagByName(devId, valObject.TagName())
		keyName := strings.TrimSpace(tag.Tag_desc) //2: 物模标识 tag_desc
		switch t.protocolInfo.TransferKey {
		case 0: //0:测点名称 tag_name
			keyName = strings.TrimSpace(tag.Tag_name)
		case 1: //1:测点地址 obj_addr
			keyName = strings.TrimSpace(tag.Obj_addr)
		}
		arr[keyName] = valObject.Val()
	}

	list := make([]interface{}, 0, len(groupByTs))

	for ts, values := range groupByTs {
		entry := make(map[string]interface{}, 2)
		entry["values"] = values
		entry["ts"] = ts
		list = append(list, entry)
	}
	devName := deviceConfInfo.Dev_name
	if t.protocolInfo.PublicAddrAsDeviceName() {
		devName = deviceConfInfo.Dev_pub_addr
	} else if t.protocolInfo.CommunicationAddrAsDeviceName() {
		devName = deviceConfInfo.Dev_comm_addr
	}

	resultMap := map[string]interface{}{devName: list}

	return resultMap
}
